/*
 * Copyright 2011-2012 by The Regents of the University of California Licensed
 * under the Apache License, Version 2.0 (the "License"); you may not use this
 * file except in compliance with the License. you may obtain a copy of the
 * License from
 * 
 * http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations under
 * the License.
 */
package edu.ucsd.db.bassarisk.storage.loading;

import java.nio.ByteBuffer;

import edu.ucsd.db.bassarisk.data.values.IntType;
import edu.ucsd.db.bassarisk.storage.common.ColumnStore;

/**
 * What is in the column header?
 * <ul>
 * <li>2 bytes :: column index version
 * <li>4 bytes :: maximum value in column
 * <li>2 bytes :: number of cards in index
 * <li>4 bytes :: number of drawers in column
 * <li>8 bytes :: total number of entries
 * <li>TOTAL = 20 bytes
 * </ul>
 * 
 * <p>
 * The 2 bytes for number of index cards limits column to 2^22 drawers, or about
 * 128GB of data using this style header. But by limiting the column header to
 * one drawer, we limit to 2**13 drawers, or about 256MB of data.
 * 
 * <p>
 * What is in the column index? (Each per drawer!)
 * <ul>
 * <li>4 bytes :: fist value on card
 * <li>0 bytes :: first entry on card (can be computed for fixed-width)
 * </ul>
 */
public class FixedLengthAscendingIntegerColumnIndexer extends
    FixedLengthAscendingIntegerSegmentIndexer implements IColumnIndexer<IntType> {

  private static final short columnIndexVersion = 2;

  private static final int BYTES_PER_ENTRY = 8;
  private static final int BYTES_IN_HEADER = 20;
  private static final int ENTIRES_IN_FULL_DRAWER = 8064;
  
  private int columnMaxValue;

  private final int[] drawerAddress;

  public FixedLengthAscendingIntegerColumnIndexer() {
    super();
    this.segmentMinValues = new int[ColumnStore.BYTES_IN_DRAWER / BYTES_PER_ENTRY];
    this.segmentValueCounts = new int[ColumnStore.BYTES_IN_DRAWER / BYTES_PER_ENTRY];
    this.drawerAddress = new int[ColumnStore.BYTES_IN_DRAWER / BYTES_PER_ENTRY];
  }

  @Override
  protected int getBytesForHeader() {
    return BYTES_IN_HEADER;
  };

  @Override
  protected int getBytesPerEntry() {
    return BYTES_PER_ENTRY;
  };

  @Override
  public void indexDrawer(final ColumnSegmentInfo<IntType> drawerInfo,
                          final int address) {
    this.segmentMinValues[this.numIndexedSegments] =
        intManip.getIntegerValue(drawerInfo.getFirstValue());
    this.columnMaxValue = intManip.getIntegerValue(drawerInfo.getLastValue());
    this.segmentValueCounts[this.numIndexedSegments] =
        (int) drawerInfo.getEntryCount();
    this.drawerAddress[this.numIndexedSegments] = address;
    this.numIndexedSegments++;
  }

  @Override
  protected void
      writeHeaderValues(final ByteBuffer drawer, final int headerEndOffset) {
    // position at start of header information
    drawer.position(headerEndOffset - BYTES_IN_HEADER);
    long totalValues = 0;
    for (int i = 0; i < this.numIndexedSegments; i++) {
      totalValues += this.segmentValueCounts[i];
      // all drawers must be full except the last one.
      assert (this.segmentValueCounts[i] == ENTIRES_IN_FULL_DRAWER || (i + 1) == this.numIndexedSegments);
    }
    drawer.putLong(totalValues);
    drawer.putInt(this.numIndexedSegments);
    drawer.putShort((short) this.cardsRequired());
    drawer.putInt(this.columnMaxValue);
    drawer.putShort(FixedLengthAscendingIntegerColumnIndexer.columnIndexVersion);
  }

  @Override
  public int writeIndex(final ByteBuffer drawer, final int startCardIndex) {
    int cardsRequired = this.cardsRequired();
    int lastCardIndex = startCardIndex + cardsRequired - 1;
    assert lastCardIndex < ColumnStore.CARDS_IN_DRAWER;

    int headerEndOffset = (lastCardIndex + 1) * ColumnStore.BYTES_IN_CARD;
    int spaceRequired = cardsRequired * ColumnStore.BYTES_IN_CARD;
    drawer.position(headerEndOffset - spaceRequired);

    int i = 0;
    for (; i < this.numIndexedSegments; ++i) {
      drawer.putInt(this.segmentMinValues[i]);
    }
    for (i = 0; i < this.numIndexedSegments; ++i) {
      drawer.putInt(this.drawerAddress[i]);
    }

    this.writeHeaderValues(drawer, headerEndOffset);
    return cardsRequired;
  }

}
